from flask import Blueprint, render_template, jsonify, request
from flask_login import login_required
from user.role_check import get_all_cn
from utilities.cache_config import cache
from utilities.hdfs_helper import get_hdfs_path
from config import kafka_url, hue_hive, hue_hbase
import json
from utilities.web_socket import socketio
from flask_socketio import join_room, emit, disconnect
from user.role_check import check_role
import subprocess
from utilities.time_format import datetime_timestamp
query = Blueprint('query', __name__)


@query.route('/data_query')
@login_required
@check_role('3.1')
def data_query():
    return render_template('data_query/data_query.html', hue_hive=hue_hive, hue_hbase=hue_hbase)


@query.route('/hdfs_nodes')
@login_required
def get_nodes():
    hdfs_path = request.args['node']

    @cache.cached(timeout=1800, key_prefix=hdfs_path)
    def get_hdfs_files():
        dataresult = get_hdfs_path(path=hdfs_path)
        return json.dumps(dataresult)

    return get_hdfs_files()


@query.route('/data_query_hdfs')
@login_required
@check_role('3.2')
def data_query_hdfs():
    return render_template('data_query/data_query_hdfs.html')


@query.route('/data_query_kafka')
@login_required
def data_query_kafka():
    # if '3.3' in get_all_cn()[current_user.get_cn()] or 'all' in get_all_cn()[current_user.get_cn()]:
    return render_template('data_query/data_query_kafka.html')
    # else:
    #     abort(401)


def background_thread(room, topic, flag, filterstring):
    from pykafka import KafkaClient
    import time
    oldtime = time.time()
    client = KafkaClient(hosts=kafka_url)

    if str(topic) not in client.topics:
        socketio.emit('my_response',
                      {'data': 'warn:kafka has no this topic!'},
                      namespace='/data_query_kafka', room=room)
        return
    topic = client.topics[str(topic)]
    reset = -2 if flag else -1
    consumer = topic.get_simple_consumer(
        auto_offset_reset=reset,
        consumer_timeout_ms=300000
        # reset_offset_on_start=False
    )
    filterstring = filterstring.encode("utf-8")
    for message in consumer:
        if filterstring not in message.value:
            print type(filterstring), type(message.value)
            continue
        socketio.emit('my_response',
                      {'data': 'offset:' + str(message.offset) + ', message:' + message.value},
                      namespace='/data_query_kafka', room=room)
        print message.offset
        time.sleep(0.1)
        if time.time() - oldtime > 300:
            break
    print 'over'


@socketio.on('my_event', namespace='/data_query_kafka')
def test_message(message):
    room = message['room']
    topic = message['data']
    flag = message['frombegin']
    filterstring = message['filterstring']
    join_room(room)
    emit('my_response', {'data': "-----------------------------------" + message['info'] + ' ' + message['data']})
    socketio.start_background_task(target=background_thread, room=room, topic=topic, flag=flag,
                                   filterstring=filterstring)


@login_required
@socketio.on('disconnect_request', namespace='/data_query_kafka')
def disconnect_request(message):
    socketio.close_room(room=message['room'])
    emit('my_response',
         {'data': '----------------------------------- disconnected!'})
    disconnect()


@login_required
@socketio.on('disconnect', namespace='/data_query_kafka')
def test_disconnect():
    print('client disconnected', request.sid)


def get_hbase_result(command):
    subp = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE)
    result = None
    while True:
        c = subp.stdout.readline()
        if 'error---' in c:
            result = 'error', c[c.index('error---') + 8:]
        elif 'ok---' in c:
            result = 'ok', c[c.index('ok---') + 5:]
        if not c:
            break
    return result


@query.route('/data_query_user_info')
def data_query_user_info():
    return render_template('data_query/data_query_user_info.html')


@query.route('/data_query_get_user_info')
def data_query_get_user_info():
    argdict = request.args
    result = {'status': None}
    pre_comm = 'java -jar D:\work\dataportal_new\data_query\hbase_query_tool-1.0-SNAPSHOT-jar-with-dependencies.jar '
    date_str=datetime_timestamp(argdict['date'])
    comm = argdict['key'] + ' ' + argdict['value']+" "+date_str
    re = get_hbase_result(pre_comm + comm)
    result['status'] = re[0]
    result['message'] = re[1] if re[0] == 'error' else json.loads(re[1])
    return jsonify(result)
